package com.study.flink.datastream

import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}

/**
  * Flink KeyBy Demo
  *
  * @author stephen
  * @create 2019-05-25 12:29
  * @since 1.0.0
  */
object FlinkKeyByDemo {

  def main(args: Array[String]): Unit = {
    // 1、执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2、指定数据源
    import org.apache.flink.api.scala._
    val dataStream = env.fromElements(("a", 1), ("b", 2), ("c", 3), ("a", 1), ("a", 1), ("b", 1))

    // 3、对数据进行操作
    val keyedStream: KeyedStream[(String, Int), String] = dataStream.keyBy(_._1)
    val resultStream: DataStream[(String, Int)] = keyedStream.reduce((t1, t2) => (t1._1, t1._2 + t2._2))
    //val resultStream = keyedStream.sum(1)
    //val resultStream = keyedStream.max(1)
    //val resultStream = keyedStream.maxBy(1)

    // 4、指定数据输出位置
    resultStream.print()

    // 5、触发任务执行
    env.execute("Flink KeyBy Demo")
  }
}
